package org.sunyaxing.transflow.api.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 利用此工具连接 transflow 服务端并传输数据
 */
public class DataTransor {
    private static final Logger log = LoggerFactory.getLogger(DataTransor.class);
    private final int maxConnectTimes = 10;
    private final String host;
    private final Integer port;
    private EventLoopGroup group;
    private Channel channel;

    private DataTransor(Builder builder) {
        host = builder.host;
        port = builder.port;
    }

    public void connect() {
        group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        try {
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            // HTTP协议编解码器
                            pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Unpooled.copiedBuffer("\n".getBytes())));
                            // 聚合HTTP报文为FullHttpRequest/Response
                            pipeline.addLast(new StringDecoder());
                            // WebSocket协议处理器（自动处理握手和协议升级）
                            pipeline.addLast(new StringEncoder());
                            // 自定义业务处理器
                            pipeline.addLast(new NettyTcpClientHandler(DataTransor.this));
                        }
                    });
            final AtomicInteger connectTimes = new AtomicInteger(0);
            ChannelFuture channelFuture = bootstrap.connect(host, port);
            channelFuture.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    // 连接成功时打印日志
                    log.info("成功连接到端口: {}", port);
                } else {
                    // 连接失败时打印日志
                    log.error("连接到服务器时出错，端口: {}", port);
                    // 连接失败时，使用 eventLoop 安排重连任务，60 秒后重连
                    if (connectTimes.get() < maxConnectTimes) {
                        connectTimes.incrementAndGet();
                        future.channel().eventLoop().schedule(() -> {
                            // 重连逻辑，再次尝试连接
                            bootstrap.connect(host, port);
                        }, 60, TimeUnit.SECONDS);
                    } else {
                        log.error("已达到最大连接次数，停止重连，端口: {}", port);
                    }
                }
            });
            channel = channelFuture.channel();
        } catch (Exception e) {
            log.error("e: ", e);
        }
    }

    public void send(String msg) {
        if (channel != null) {
            channel.writeAndFlush(msg + "\n");
        } else {
            log.warn("通道未建立，请检查");
        }
    }

    public void disconnect() {
        group.shutdownGracefully();
    }

    public static final class Builder {
        private String host;
        private Integer port;

        private Builder() {
        }

        public static Builder builder() {
            return new Builder();
        }

        public Builder host(String val) {
            host = val;
            return this;
        }

        public Builder port(Integer val) {
            port = val;
            return this;
        }

        public DataTransor build() {
            return new DataTransor(this);
        }
    }

}
